package com.distributer.message.producer;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import com.distributer.message.model.Message;

/**
 * 暂时每个队列只使用一个线程进行处理，可以根据消息数量来调整线程数，优化成使用线程池。
 * @author yilin
 *
 */
public class ProducerThread implements Runnable{

	private static final Logger LOGGER = LoggerFactory.getLogger(ProducerThread.class);
	
	private ArrayBlockingQueue<Message> queue;
	
	private KafkaTemplate<String, String> kafkaTemplate;
	/**使用8的原因是通过测试key更平均的将每个消息平均的分布到不同的分区中达到各个分区数据的平衡*/
	private static final int KAFKA_DEFAULT_PATITION = 8;
	
	public ProducerThread(ArrayBlockingQueue<Message> queue,KafkaTemplate<String, String> kafkaTemplate){
		this.queue = queue;
		this.kafkaTemplate = kafkaTemplate;
	}
	
	@Override
	public void run() {
		while (!Thread.interrupted()) {
			try {
				Thread.sleep(1);
				Message message = queue.take();
				int patition = getPatitionByKey(message.getMessageSquece());
				LOGGER.info("消息序列号： "+message.getMessageSquece()+ ",分区编号： "+ patition);
				ProducerRecord<String,String> record = new ProducerRecord<String, String>(message.getNotifyTopic(), patition, message.getMessageSquece(), message.getNotifyData());
				kafkaTemplate.send(record);
			} catch (InterruptedException e) {
				LOGGER.info("-----------InterruptedException----------");
				LOGGER.info("未发送的消息数量： "+ queue.size());
				LOGGER.info("线程："+Thread.currentThread().getName()+"停止执行消息处理！");
			}catch (Exception e) {
				LOGGER.info("---------kafka stop error---------");
			}
		}
	}

	private static int getPatitionByKey(String key){
		int hashValue = key.hashCode() >>> 16;
		int patitionIndex = (KAFKA_DEFAULT_PATITION - 1) & hashValue;
		return patitionIndex;
	}
	
	public static void main(String[] args) {
		Random random = new Random();
		for (int i = 0; i < 20; i++) {
			String str = UUID.randomUUID().toString().replace("-", "").substring(0, 4);
			
			System.out.println(str +" : "+getPatitionByKey(str+""));
		}
		
		
	}
}
